package com.hqbzl.chat;

import com.hqbzl.chat.constant.Constants;
import com.hqbzl.chat.handler.RpcResponseHandler;
import com.hqbzl.chat.message.RpcRequestMessage;
import com.hqbzl.chat.model.UserInfo;
import com.hqbzl.chat.protocol.MessageCodec;
import com.hqbzl.chat.protocol.ProcotolFramDecoder;
import com.hqbzl.chat.service.UserService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Proxy;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class RpcClientManager {
    static AtomicInteger SEQUENCEIDGEN = new AtomicInteger();

    private static Channel channel;
    private static final Object object = new Object();

    public static void main(String[] args) {
//        final RpcRequestMessage build = RpcRequestMessage.builder()
//                .interfaceName("com.hqbzl.chat.service.UserService")
//                .methodName("say")
//                .parameterTypes(new Class[]{String.class, String.class})
//                .parameterValus(new Object[]{"1", "2"})
//                .build();
//        final Channel channel = getChannel();
//        log.debug("{}", channel);
//        channel.writeAndFlush(build);
        final UserService service = getProxyService(UserService.class);
//        final String say1 = service.say("a1", "b1");
//        final String say2 = service.say("a2", "b2");
//        final String say3 = service.say("a3", "b3");
//        System.out.println("say1 = " + say1);
        final UserInfo userInfo = service.info("李四");
        System.out.println("userInfo = " + userInfo);
    }

    public static <T> T getProxyService(Class<T> serviceClass) {
        final ClassLoader classLoader = serviceClass.getClassLoader();
        final Class[] interfaceClass = new Class[]{serviceClass};
        final Object o = Proxy.newProxyInstance(classLoader, interfaceClass, (proxy, method, args) -> {
            final int sequenceId = SEQUENCEIDGEN.incrementAndGet();
            // 1、将方法调用转为消息对象
            final RpcRequestMessage rpc = RpcRequestMessage
                    .builder()
                    .sequenceId(sequenceId)
                    .interfaceName(serviceClass.getName())
                    .methodName(method.getName())
                    .parameterTypes(method.getParameterTypes())
                    .parameterValus(args)
                    .returnType(method.getReturnType())
                    .build();
            // 2、将消息对象发出
            log.debug("getProxyService====================>: {}", rpc.toString());
            // 指定promise对象接收结果的线程
            getChannel().writeAndFlush(rpc);
            // 3、处理返回值 准备一个promise对象来接收结果， 参数表明接收结果的线程
            DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
            Constants.PROMISE_MAP.put(sequenceId, promise);
            // 4、等结果，await不会抛异常
            final Promise<?> await = promise.await();
            if (await.isSuccess()) {
                // 有结果正常响应
                final Object now = promise.getNow();
                return now;
            } else {
                throw new RuntimeException(promise.cause());
            }
        });
        return (T) o;
    }

    public static Channel getChannel() {
        if (channel != null) {
            return channel;
        }
        synchronized (object) {
            if (channel == null) {
                synchronized (object) {
                    initChannel();
                }
            }
        }
        return channel;
    }

    public static void initChannel() {
        final LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        final MessageCodec MESSAGE_CODEC = new MessageCodec();
        final NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            final Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(new ProcotolFramDecoder());
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    ch.pipeline().addLast(new RpcResponseHandler());
                    // 写空闲比服务器要小
                    ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
                    // 即可以同时作为入栈和出栈处理器
                    //ch.pipeline().addLast(new HeartBeatHandler());
                    //ch.pipeline().addLast(new PingMessageResponseHandler());
                }
            });
            final ChannelFuture channelFuture = bootstrap.connect("localhost", 7788).sync();
            // 连接创建好
            channel = channelFuture.sync().channel();
            // 阴赛方法，此方法不会返回
            //channel.closeFuture().sync();
            channel.closeFuture().addListener(x -> {
                group.shutdownGracefully();
            });
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("chnnal crate = " + "chnnal crate");
    }
}
